package oc.tm.sg.core.rpc;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import oc.tm.sg.core.Constants;
import oc.tm.sg.core.monitor.Statistics;
import oc.tm.sg.core.rpc.consumer.RpcConsumerFactory;
import oc.tm.sg.core.util.DateUitls;
import oc.tm.sg.core.util.NetUtils;
import oc.tm.sg.core.util.StringUtils;

import org.zbus.broker.BrokerConfig;
import org.zbus.kit.JsonKit;
import org.zbus.kit.log.Logger;
import org.zbus.net.Sync.ResultCallback;
import org.zbus.net.http.Message;
import org.zbus.net.http.Message.MessageInvoker;
import org.zbus.rpc.JsonRpcCodec;
import org.zbus.rpc.RpcCodec;
import org.zbus.rpc.RpcCodec.Request;
import org.zbus.rpc.RpcCodec.Response;
import org.zbus.rpc.RpcConfig;
import org.zbus.rpc.RpcException;

public class SgRpcInvoker{  
	private static final Logger log = Logger.getLogger(SgRpcInvoker.class); 
	
	private MessageInvoker messageInvoker; 
	private RpcCodec codec; 
	private BrokerConfig brokerConfig;
	
	private String module = ""; 
	private String encoding = "UTF-8";
	private int timeout = 10000;  
	private boolean verbose = false;

	private Map<String,String> param;
	private RpcConsumerFactory consumer;
	
	//并发计数
	private final ConcurrentMap<String, AtomicInteger> concurrents = new ConcurrentHashMap<String, AtomicInteger>();

	public SgRpcInvoker(MessageInvoker messageInvoker){
		this(messageInvoker, new JsonRpcCodec(), null);
	}
	
	public SgRpcInvoker(MessageInvoker messageInvoker, RpcCodec codec){
		this(messageInvoker, codec, null);
	}
	
	public SgRpcInvoker(MessageInvoker messageInvoker, RpcConfig config){
		this(messageInvoker, new JsonRpcCodec(), config);
	}
	
	public SgRpcInvoker(MessageInvoker messageInvoker, RpcCodec codec, RpcConfig config){ 
		this.messageInvoker = messageInvoker; 
		this.codec = codec;
		if(config != null){
			this.module = config.getModule();
			this.timeout = config.getTimeout(); 
			this.encoding = config.getEncoding();
			this.verbose = config.isVerbose();
		}
	}
	
	public SgRpcInvoker(BrokerConfig brokerConfig, MessageInvoker messageInvoker, RpcConfig config) {
		this(messageInvoker, new JsonRpcCodec(), config);
		this.brokerConfig=brokerConfig;
	}

	public void init(RpcConsumerFactory consumer,Map<String,String> param){
		this.consumer=consumer;
		this.param=param;
	}
	
	public <T> T invokeSync(Class<T> resultClass, String method, Object... args){
		Request request = new Request()
			.module(module)
			.method(method)  
			.params(args)
			.encoding(encoding);

		return invokeSync(resultClass, request);
	}
	
	public <T> T invokeSync(Class<T> resultClass, String method, Class<?>[] paramTypes, Object... args){
		Request request = new Request()
			.module(module)
			.method(method)
			.paramTypes(paramTypes)
			.params(args)
			.encoding(encoding);
	
		return invokeSync(resultClass, request);
	} 
	
	public <T> T invokeSync(Class<T> resultClass, Request request){
		Response resp = invokeSync(request);
		try {
			@SuppressWarnings("unchecked")
			T res = (T)codec.convert(extractResult(resp), resultClass);
			return res;
		} catch (ClassNotFoundException e) { 
			throw new RpcException(e.getMessage(), e.getCause());
		}
	}
	
	
	public Object invokeSync(String method, Object... args) {	
		return invokeSync(method, null, args);
	}  
	
	public Object invokeSync(String method, Class<?>[] types, Object... args) {	
		Request req = new Request()
			.module(module)
			.method(method) 
			.paramTypes(types)
			.params(args)
			.encoding(encoding); 
		 
		Response resp = invokeSync(req);
		return extractResult(resp);
	} 
	
	public Response invokeSync(Request request){
		long start=0,data_length=0;
		Message msgReq= null, msgRes = null;
		try {
			data_length=JsonKit.toJson(request.getParams()).length();
			msgReq = codec.encodeRequest(request);
			
			msgReq.setHead(Constants.HOSTNAME, NetUtils.getLocalHostName());
			msgReq.setHead(Constants.IP, NetUtils.getLocalIp());
			msgReq.setHead(Constants.PID, NetUtils.getPid());
			msgReq.setHead(Constants.SERVICE, module);
			msgReq.setHead(Constants.METHOD, request.getMethod());
			
			if(isVerbose()){
				log.info("[REQ]: %s", msgReq);
			} 
			
			start = System.currentTimeMillis();
			msgRes = messageInvoker.invokeSync(msgReq, this.timeout); 
			
			/** 【<<埋点一>>】:组装统计数据 **/
			assemble(request,msgReq, start,data_length,true);
			
			if(isVerbose()){
				log.info("[REP]: Time cost=%dms\n%s",(System.currentTimeMillis()-start), msgRes);
			} 
		} catch (Throwable t) {
			/** 【<<埋点二>>】:组装统计数据 **/
			assemble(request,msgReq, start,data_length,false);
			
			throw new RpcException(t.getMessage(), t);
		}
		
		if (msgRes == null) { 
			/** 【<<埋点三>>】:组装统计数据 **/
			assemble(request,msgReq, start,data_length,false);
			
			String errorMsg = String.format("module(%s)-method(%s) request timeout\n%s",  module, request.getMethod(), msgReq.toString());
			throw new RpcException(errorMsg);
		}
		
		return codec.decodeResponse(msgRes);
	}

	
	public <T> void invokeAsync(final Class<T> clazz, Request request,  final ResultCallback<T> callback){
		invokeAsync(request, new ResultCallback<Response>() { 
			@Override
			public void onReturn(Response resp) {  
				Object netObj = extractResult(resp);
				try {
					@SuppressWarnings("unchecked")
					T target = (T) codec.convert(netObj, clazz);
					callback.onReturn(target);
				} catch (ClassNotFoundException e) { 
					throw new RpcException(e.getMessage(), e.getCause());
				}
			}
		});
	}
	
	public void invokeAsync(Request request, final ResultCallback<Response> callback){ 
		final long start = System.currentTimeMillis();
		final Message msgReq = codec.encodeRequest(request); 
		if(isVerbose()){
			log.info("[REQ]: %s", msgReq);
		}  
		try {
			messageInvoker.invokeAsync(msgReq, new ResultCallback<Message>() {
				@Override
				public void onReturn(Message result) { 
					if(isVerbose()){
						long end = System.currentTimeMillis();
						log.info("[REP]: Time cost=%dms\n%s",(end-start), result); 
					} 
					Response resp = codec.decodeResponse(result);
					if(callback != null){
						callback.onReturn(resp);
					}
				}
			});
		} catch (IOException e) {
			throw new RpcException(e.getMessage(), e);
		}  
	}
	
	private Object extractResult(Response resp){
		if(resp.getStackTrace() != null){
			Throwable error = resp.getError();
			if(error != null){
				if(error instanceof RuntimeException){
					throw (RuntimeException)error;
				}
				throw new RpcException(error.getMessage(), error.getCause()); 
			} else {
				throw new RpcException(resp.getStackTrace());
			}
		} 
		return resp.getResult();
	}
	
	public String getEncoding() {
		return encoding;
	}

	public void setEncoding(String encoding) {
		this.encoding = encoding;
	}

	public String getModule() {
		return module;
	}

	public void setModule(String module) {
		this.module = module;
	}
	
	public SgRpcInvoker module(String module) {
		this.module = module;
		return this;
	}

	public int getTimeout() {
		return timeout;
	}

	public void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	public boolean isVerbose() {
		return verbose;
	}

	public void setVerbose(boolean verbose) {
		this.verbose = verbose;
	} 
	
	/**
	 * 收集数据组装
	 * @param msgReq
	 * @param start
	 */
	@SuppressWarnings("deprecation")
	public Statistics assemble(Request request,Message msg,long start,long data_length,boolean isSuccessed){
		long elapsed=System.currentTimeMillis()-start;
		
		Statistics statistics=new Statistics();
		statistics.setId(UUID.randomUUID().toString());
		
		statistics.setService(module);//服务接口
		String group=param.get(Constants.GROUP_KEY);
		statistics.setGroup(StringUtils.isBlank(group)?"":group);//服务分组
		String version=param.get(Constants.VERSION_KEY);
		statistics.setVersion(StringUtils.isBlank(version)?Constants.DEFAULT_VERSION:version);//服务版本
		
		statistics.setMethod(request.getMethod());//需要调度的方法
		statistics.setApplication(consumer.getGovernConf().application());//应用名称
		statistics.setConsumer(NetUtils.getLocalIp());//当前消费者IP
		statistics.setProvider(NetUtils.getIpByAddress(brokerConfig.getServerAddress()));//被调度的IP
		
		statistics.setData_length(data_length);//数据长度
		statistics.setType(Constants.CONSUMER);//数据统计来源
		statistics.setInvoke_date(DateUitls.timestamp2Date(start, "yyyy-MM-dd"));//调度日期
		statistics.setInvoke_time(start);//调度时间戳
		
		statistics.setSuccess(isSuccessed?1:0);//成功次数
		statistics.setFailure(isSuccessed?0:1);//失败次数
		statistics.setElapsed(elapsed);//耗时
		statistics.setConcurrent(getConcurrent(msg).get());//并发数
		statistics.setMax_elapsed(statistics.getElapsed());//最大耗时
		statistics.setMax_concurrent(statistics.getConcurrent());//最大并发数
		
		//内存收集
		consumer.getMemoryMonitor().collect(statistics);
		
		return statistics;
	}
	
	// 获取并发计数器
    private AtomicInteger getConcurrent(Message msg) {
        String key = msg.getHead(Constants.SERVICE) + "." + msg.getHead(Constants.METHOD);
        AtomicInteger concurrent = concurrents.get(key);
        if (concurrent == null) {
            concurrents.putIfAbsent(key, new AtomicInteger());
            concurrent = concurrents.get(key);
        }
        return concurrent;
    }
	
}
